package com.pw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.app.BaseAppV2;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.VisitorState;
import com.pw.gmall.realtime.utils.AtguiguUtils;
import com.pw.gmall.realtime.utils.FlinkSinkUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashMap;

/**
 * @Author: linux_future
 * @since: 2022/3/25
 **/
public class DwsVisitorApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsVisitorApp().init(2020, 1, "dwsVisitorApp", "dwsVisitorApp",
                Constant.TOPIC_DWD_PAGE_LOG, Constant.TOPIC_DWM_UV, Constant.TOPIC_DWM_UJ);
    }

    @Override
    protected void handler(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> stream) {
        //处理数据流并连接在一起
        DataStream<VisitorState> unionStream = etlStream(stream);
        //unionStream.print("union:");
        // 处理数据，分维度，做累加
        SingleOutputStreamOperator<VisitorState> reduceStream = dealStream(unionStream);
        //写入clickhouse
        writeToClickhouse(reduceStream);


    }

    private void writeToClickhouse(SingleOutputStreamOperator<VisitorState> stream) {

        stream.addSink(FlinkSinkUtils.getClickhouseSink(Constant.CLICKHOUSE_DB, Constant.CLICKHOUSE_TABLE_VISITOR_STATS, VisitorState.class));
    }

    private SingleOutputStreamOperator<VisitorState> dealStream(DataStream<VisitorState> stream) {
        //可能会出现，有些数据没有及时到窗口的情况
        OutputTag<VisitorState> outputTag = new OutputTag<VisitorState>("timeOut") {
        };
        WatermarkStrategy<VisitorState> wms = WatermarkStrategy.<VisitorState>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((bean, ts) -> bean.getTs());
        //选取4个维度做为key:版本, 渠道, 地区, 新老用户标识
        SingleOutputStreamOperator<VisitorState> reduceStream = stream.assignTimestampsAndWatermarks(wms)
                .keyBy(bean -> bean.getVc() + "_" + bean.getCh() + "_" + bean.getAr() + "_" + bean.getIs_new())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(outputTag)
                .reduce(new ReduceFunction<VisitorState>() {
                    @Override
                    public VisitorState reduce(VisitorState v1, VisitorState v2) throws Exception {

                        //累加
                        v1.setUv_ct(v1.getUv_ct() + v2.getUv_ct());
                        v1.setPv_ct(v1.getPv_ct() + v2.getPv_ct());
                        v1.setSv_ct(v1.getSv_ct() + v2.getSv_ct());
                        v1.setUj_ct(v1.getUj_ct() + v2.getUj_ct());
                        v1.setDur_sum(v1.getDur_sum() + v2.getDur_sum());
                        return v1;
                    }
                }, new ProcessWindowFunction<VisitorState, VisitorState, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context ctx, Iterable<VisitorState> elements, Collector<VisitorState> out) throws Exception {
                        VisitorState vs = elements.iterator().next();
                        String startTime = AtguiguUtils.toDateTime(ctx.window().getStart());
                        String endTime = AtguiguUtils.toDateTime(ctx.window().getEnd());
                        vs.setStt(startTime);
                        vs.setEdt(endTime);
                        vs.setTs(System.currentTimeMillis());
                        out.collect(vs);
                    }
                });
        return reduceStream;

    }

    private DataStream<VisitorState> etlStream(HashMap<String, DataStreamSource<String>> stream) {
        //1. pv和持续访问时间
        SingleOutputStreamOperator<VisitorState> pageStream = stream.get(Constant.TOPIC_DWD_PAGE_LOG).map(obj -> {
            JSONObject json = JSON.parseObject(obj);
            JSONObject common = json.getJSONObject("common");
            String vc = common.getString("vc");
            String ch = common.getString("ch");
            String ar = common.getString("ar");
            String is_new = common.getString("is_new");
            Long ts = json.getLong("ts");
            Long duringTime = json.getJSONObject("page").getLong("during_time");

            return new VisitorState("", "", vc, ch, ar, is_new,
                    0L, 1L, 0L, 0L, duringTime, ts);
        });
        //uv> {"common":{"ar":"110000","uid":"2","os":"Android 11.0","ch":"web","is_new":"1","md":"Honor 20s","mid":"mid_42","vc":"v2.1.134","ba":"Honor"},"page":{"page_id":"good_detail","item":"4","during_time":12244,"item_type":"sku_id","last_page_id":"home","source_type":"query"},"displays":[{"display_type":"query","item":"5","item_type":"sku_id","pos_id":3,"order":1},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":2,"order":2},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":2,"order":3},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":1,"order":4},{"display_type":"promotion","item":"5","item_type":"sku_id","pos_id":4,"order":5},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":1,"order":6},{"display_type":"promotion","item":"8","item_type":"sku_id","pos_id":4,"order":7},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":4,"order":8},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":2,"order":9}],"actions":[{"item":"4","action_id":"favor_add","item_type":"sku_id","ts":1616664673122}],"ts":1616664667000}
        SingleOutputStreamOperator<VisitorState> uvStream = stream.get(Constant.TOPIC_DWM_UV).map(obj -> {
            JSONObject json = JSON.parseObject(obj);
            JSONObject common = json.getJSONObject("common");
            String vc = common.getString("vc");
            String ch = common.getString("ch");
            String ar = common.getString("ar");
            String is_new = common.getString("is_new");
            Long ts = json.getLong("ts");

            return new VisitorState("", "", vc, ch, ar, is_new,
                    1L, 0L, 0L, 0L, 0L, ts);
        });
        //u j> {"common":{"ar":"440000","uid":"22","os":"Android 10.0","ch":"oppo","is_new":"0","md":"Huawei Mate 30","mid":"mid_46","vc":"v2.1.134","ba":"Huawei"},"page":{"page_id":"home","during_time":19307},"displays":[{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":4,"order":1},{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":4,"order":2},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":3,"order":3},{"display_type":"promotion","item":"7","item_type":"sku_id","pos_id":4,"order":4},{"display_type":"promotion","item":"7","item_type":"sku_id","pos_id":5,"order":5},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":3,"order":6},{"display_type":"query","item":"2","item_type":"sku_id","pos_id":1,"order":7},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":5,"order":8},{"display_type":"promotion","item":"2","item_type":"sku_id","pos_id":2,"order":9},{"display_type":"promotion","item":"4","item_type":"sku_id","pos_id":1,"order":10},{"display_type":"promotion","item":"4","item_type":"sku_id","pos_id":2,"order":11},{"display_type":"promotion","item":"3","item_type":"sku_id","pos_id":3,"order":12}],"ts":1616664822000}
        SingleOutputStreamOperator<VisitorState> ujStream = stream.get(Constant.TOPIC_DWM_UJ).map(obj -> {
            JSONObject json = JSON.parseObject(obj);
            JSONObject common = json.getJSONObject("common");
            String vc = common.getString("vc");
            String ch = common.getString("ch");
            String ar = common.getString("ar");
            String is_new = common.getString("is_new");
            Long ts = json.getLong("ts");

            return new VisitorState("", "", vc, ch, ar, is_new,
                    0L, 0L, 0L, 1L, 0L, ts);
        });
        SingleOutputStreamOperator<VisitorState> svStream = stream.get(Constant.TOPIC_DWD_PAGE_LOG)
                .flatMap(new FlatMapFunction<String, VisitorState>() {
                    @Override
                    public void flatMap(String json, Collector<VisitorState> out) throws Exception {
                        JSONObject obj = JSON.parseObject(json);
                        String lastPageId = obj.getJSONObject("page").getString("last_page_id");

                        if (lastPageId == null || lastPageId.length() == 0) {
                            JSONObject common = obj.getJSONObject("common");
                            String vc = common.getString("vc");
                            String ch = common.getString("ch");
                            String ar = common.getString("ar");
                            String isNew = common.getString("is_new");
                            Long ts = obj.getLong("ts");
                            VisitorState vs = new VisitorState(
                                    "", "",
                                    vc, ch, ar, isNew,
                                    0L, 0L, 1L, 0L, 0L,
                                    ts
                            );

                            out.collect(vs);
                        }
                    }
                });
/**
 pageStream.print("page");
 uvStream.print("uv");
 ujStream.print("uj");
 svStream.print("sv");
 */
        DataStream<VisitorState> unionStream = pageStream.union(uvStream, ujStream, svStream);
        return unionStream;
    }
}
